//package com.kingsoft.dc.khaos.module.spark.source
//
//import com.kingsoft.dc.khaos.KhaosContext
//import com.kingsoft.dc.khaos.metadata.{Dependency, KhaosStructField}
//import com.kingsoft.dc.khaos.module.spark.constants.HDFSEnum
//import com.kingsoft.dc.khaos.module.spark.metadata.source.HdfsConfig
//import com.kingsoft.dc.khaos.module.spark.util.DataframeUtils
//import com.kingsoft.dc.khaos.util.Logging
//import org.apache.spark.rdd.RDD
//import org.apache.spark.sql.types.{StructField, StructType}
//import org.apache.spark.sql.{DataFrame, Row}
//import org.json4s.DefaultFormats
//import org.json4s.jackson.JsonMethods.parse
//
//import scala.collection.mutable.ArrayBuffer
//
//
///**
//  * create by yansu on 2019/09/11 15:29
//  */
//class DFSSource extends SourceStrategy with Logging {
//  /** 数据抽取 */
//  override def source(kc: KhaosContext,
//                      module_id: String,
//                      config: String,
//                      dependence: Dependency): DataFrame = {
//
//    var resDF: DataFrame = null
//
//    implicit val formats = DefaultFormats
//    val hdfsConf = parse(config, true).extract[HdfsConfig]
//
//    val sparkSession = kc.sparkSession
//    val sourcePathStr = hdfsConf.path
//
//    if (sourcePathStr == "" || sourcePathStr == null) {
//      throw new IllegalArgumentException(s"HDFS路径不能为空!")
//    }
//    // 数据类型不为空
//    if (hdfsConf.data_type.trim.toLowerCase != "" && hdfsConf.data_type.trim.toLowerCase != null) {
//      // 根据hdfs文件格式进行读取
//      hdfsConf.data_type.trim.toLowerCase match {
//        case HDFSEnum.ORC => {
//          resDF = sparkSession.read.orc(sourcePathStr)
//        }
//        case HDFSEnum.TEXT => {
//          // 根据路径获取rdd
//          val textRDD: RDD[String] = kc.sparkSession.sparkContext.textFile(sourcePathStr)
//
//          val delimit: String = hdfsConf.delimit
//          val rdd = textRDD.map(row => {
//            var lineRow: Row = Row()
//            // TODO
//            val lines = row.split("\\" + delimit)
//            lines.foreach(field => lineRow = Row.merge(lineRow, Row(field)))
//            lineRow
//          })
//          val extrFields = hdfsConf.extract_fields.map { ef =>
//            StructField(ef.field, DataframeUtils.getDataType(ef.data_type), nullable = true)
//          }
//          val schema = StructType(extrFields)
//
//          // 构建DataFrame
//          resDF = sparkSession.createDataFrame(rdd, schema)
//        }
//        case HDFSEnum.PARQUET => {
//          resDF = sparkSession.read.parquet(sourcePathStr)
//        }
//        case _ => {
//          throw new IllegalArgumentException(s"暂不支持读取 ${hdfsConf.data_type} 类型数据!")
//        }
//      }
//    } else {
//      throw new IllegalArgumentException("数据类型不能为空!")
//    }
//    if (hdfsConf.filter != null && hdfsConf.filter != "") {
//      resDF = resDF.filter(hdfsConf.filter)
//    }
//    resDF
//  }
//
//  override def schema(kc: KhaosContext, config: String, dependence: Dependency): List[KhaosStructField] = {
//    val fieldSchema = ArrayBuffer[KhaosStructField]()
//    implicit val formats = DefaultFormats
//    val info = parse(config, true).extract[HdfsConfig]
//    val extrFields = info.extract_fields
//    for (ef <- extrFields) {
//      fieldSchema += KhaosStructField(ef.field, ef.data_type)
//    }
//    fieldSchema.toList
//  }
//}
